package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

public class DosStatusAbnorBolt extends AbstractBolt{


	private static final long serialVersionUID = 1L;		
	private Map<String,Integer> mapRecordCnt = new HashMap<String,Integer>();
	private Map<String,Integer> mapPkgSum = new HashMap<String,Integer>();
	private HashMap<String,Integer> mapStatusCnt = new HashMap<String,Integer>();
	private HashMap<String,Double> dipPInfo = new HashMap<String,Double>();
	private Map<String,String> tupleMap = new HashMap<String,String>();
	private Map<String,Integer> streampkgMap = new HashMap<String,Integer>();
	private Map<String,Integer> streamoctetsMap = new HashMap<String,Integer>();
	private Map<String,Integer> pvMap = new HashMap<String,Integer>();
	private DosStatusSliding cache;
	
	public DosStatusAbnorBolt(){		
		cache = new DosStatusSliding(3);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d","dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour", "abntype", "idcno", "sourcecountry","streamoctets","streampacket"));
		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport","accesstime","sourcearea",
				"sourcegeo","sourcecountry","destarea","destgeo","evaluate","abnormal","desc","attnum","delaytime","flag"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}

	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{	

			String dip = tuple.getStringByField("dip");			
			String sip = tuple.getStringByField("sip");
			String type =  tuple.getStringByField("type");
			int status = tuple.getIntegerByField("status");
			String dport = tuple.getStringByField("dport");
			String sport = tuple.getStringByField("sport");
			String gis = new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
			String idc = tuple.getStringByField("idcno");	
			if(tuple.contains("upstreampacket") && tuple.contains("upstreamoctets")){			
				int uppacket = tuple.getIntegerByField("upstreampacket");	
				int upstreamoctets = tuple.getIntegerByField("upstreamoctets");
				String Key = dip + "|" + dport + "|" + sip + "|" + type + "|" + idc;
				
				if(type.equals("6")){
					if(pvMap.containsKey(Key)) {				
						pvMap.put(Key, pvMap.get(Key) + 1);
					}else{
						pvMap.put(Key,1);
					}
					
					if(mapRecordCnt.containsKey(Key)) {
						mapRecordCnt.put(Key, mapRecordCnt.get(Key) + 1);
					}else{
						mapRecordCnt.put(Key,1);
					}					
					
					if(mapPkgSum.containsKey(Key)){	
						mapPkgSum.put(Key, mapPkgSum.get(Key) + uppacket);
					}else{
						mapPkgSum.put(Key, uppacket);				
					}				
					
					tupleMap.put(Key,  dip + "|" + dport + "|" + sip + "|" + sport +"|" + gis +  "|"  + type + "|"  + idc + "|"  + upstreamoctets+ "|"  + uppacket);
					
					if(status == 1){	
						if(mapStatusCnt.containsKey(Key)) {
							mapStatusCnt.put(Key, mapStatusCnt.get(Key) + 1);
						}else{
							mapStatusCnt.put(Key, 1);
						}				
					}
					
					if(streamoctetsMap.containsKey(Key)) {
						streamoctetsMap.put(Key,streamoctetsMap.get(Key) + upstreamoctets);
					}else{
						streamoctetsMap.put(Key,upstreamoctets);
					}				
					
					if(streampkgMap.containsKey(Key)) {					
						streampkgMap.put(Key,streampkgMap.get(Key) + uppacket);
					}else{
						streampkgMap.put(Key,uppacket);
					}				
				}
			}
		}
		catch(Exception ex){
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}

	//定时发送
	private void emitCountingData(OutputCollector collector) {
		try{
			HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
			SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmm");		
			Calendar c = Calendar.getInstance();
			String dataMin = sdf.format(c.getTime());
			c.add(Calendar.MINUTE, -1);// 1 minute ago
			String dateMin1 = sdf.format(c.getTime());
			c.add(Calendar.MINUTE, -1);// 2 minutes ago
			String dateMin2 = sdf.format(c.getTime());
	
			dipPInfo = new HashMap<String,Double>();
			
			
			String info = String.format("######DEBUG:DosStatusAbnorBolt MAP[%d] T[%s] T1[%s] T2[%s]", mapStatusCnt.size(),	dataMin , dateMin1 , dateMin2);
			Logger.getRootLogger().info(info);
			
			for(String dkey : mapRecordCnt.keySet()) {	
				
				//if(pvMap.get(dkey) < 30) continue;
				int pkgsum =  mapPkgSum.get(dkey);	
				if(pkgsum < 3000) continue;
				
				double bottomval =  mapRecordCnt.get(dkey);
				double topval =  0.0;
				if(mapStatusCnt.containsKey(dkey)) topval = mapStatusCnt.get(dkey);
				double retval = 0.00;
				if(topval != 0.00 && bottomval != 0.00) retval = topval/bottomval;
			
				if(retval <= 0.7){				
					Logger.getRootLogger().info(">>>key:"+dkey + "|" + dataMin + ">>>topval/bottomval:"+topval+"/"+bottomval +"=retval:" +String.valueOf(retval) + ">>>pkgsum:" + pkgsum);
					
					dipPInfo.put(dkey + "|" + dataMin, retval);//目标IP|目标端口|源IP|类型|IDC|201607271353	0.755			
					Map<Integer,HashMap<String,Double>> infoMap = cache.getTuplesWindow();			
					
					Logger.getRootLogger().info(String.format("######DEBUG:DDOS Slid Window Map[%d]", infoMap.size()));				
					
					if(infoMap != null ) {
						int num = 0;
						for(HashMap<String,Double> singleMap: infoMap.values()) {
							if(singleMap != null) {
								if(singleMap.containsKey(dkey+"|" + dateMin1) || singleMap.containsKey(dkey+"|" + dateMin2)){								
									num++;								
								}
							}
						}
						Logger.getRootLogger().info(String.format("######DosStatusAbnorBolt DEBUG:DDOS n[%d] %s" , num, dkey));
						
						if(num >= 2){
							
							if(tupleMap.containsKey(dkey)) {
								String[] tuplesp = tupleMap.get(dkey).split("\\|",-1);
								String dip = tuplesp[0];
								String dport = tuplesp[1];
								String sip = tuplesp[2];
								String sport = tuplesp[3];
								String gis = tuplesp[4];
								String type = tuplesp[5];
								String idc = tuplesp[6];
								String[] gisArray = gis.split("#");
								String destAreaName = gisArray[0];
								String destGis = gisArray[1];
								String sourceAreaName = gisArray[2];
								String sourceGis = gisArray[3];
								String sourceAreaCountry = gisArray[4];
								String sourceAreaId = gisArray[5];							
								String sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
								String sourceAreaCityId = gisArray[7];
								String sourceAreaProvinceId = gisArray[8];
								String desc = "synflood攻击";	
								
								String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()+15000));
								String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_2";	
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEIP", sip);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTPORT", dport);							
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ABRNORMAL", "5");
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTIP", dip);							
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(pkgsum));
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "EVALUATE", "40");		
								if(sourceProvinceName.equals("")) sourceProvinceName = sourceAreaCountry;
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(streamoctetsMap.get(dkey)));
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(streampkgMap.get(dkey)));		
								
								String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
						    	String rowkey3 = String.format("%s_%s", accesstime, sip);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", String.valueOf(pkgsum));
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", dport);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", type);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);
						    	
								Date current = new Date();
								SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
								SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
								SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
						    	String dateStr_h = sdf_h.format(current);
								String dateStr_d = sdf_d.format(current);
								String dateStr_hour = sdf_hour.format(current);
								
								collector.emit(new Values(dip,sip,dport,sport,dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
						    			sourceProvinceName, Integer.parseInt(streamoctetsMap.get(dkey).toString()), dateStr_hour,"SYNFLOOD" , idc, sourceAreaCountry,Long.valueOf(streamoctetsMap.get(dkey)),Long.valueOf(streampkgMap.get(dkey))));
							
							}
						}
					}
				}
			}
			if(dipPInfo.size() > 0){
				cache.addPInfo(dipPInfo);
			}
			cache.advanceHead();
			
		}catch(Exception e){
			Logger.getRootLogger().error(e);
		}finally{
			pvMap.clear();
			streamoctetsMap.clear();
			streampkgMap.clear();
			mapRecordCnt.clear();
			mapPkgSum.clear();
			mapStatusCnt.clear();
			mapRecordCnt.clear();
			dipPInfo.clear();
			tupleMap.clear();
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(tuple)) {
				emitCountingData(collector);  
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("DDosAbnorBolt throws an exception!");
		}	
	}
	

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		String dbJson = stormConf.get("dbjson").toString();
		ApolloConfig ap = new ApolloConfig(dbJson);
	}

}
